[backend:fix] Reject engine-unsupported UDF parameter and return types MCDB-100032 #closes - #155

diff --git a/singlestoredb/functions/decorator.py b/singlestoredb/functions/decorator.py
index 98a44d0d..909d741f 100644
--- a/singlestoredb/functions/decorator.py
+++ b/singlestoredb/functions/decorator.py
@@ -9,8 +9,7 @@ from typing import Union
 
 from . import utils
 from .dtypes import SQLString
-from .signature import validate_func_annotations
-from .signature import validate_type_overrides
+from .signature import get_signature
 
 
 ParameterType = Union[
@@ -97,7 +96,6 @@ def expand_types(args: Any) -> Optional[List[Any]]:
     else:
         raise TypeError(f'unrecognized type for parameter: {args}')
 
-    validate_type_overrides(expanded)
     return expanded
 
 
@@ -123,48 +121,31 @@ def _func(
         ).items() if v is not None
     }
 
-    # No func was specified, this is an uncalled decorator that will get
-    # called later, so the wrapper much be created with the func passed
-    # in at that time.
-    if func is None:
-        def decorate(func: UDFType) -> UDFType:
-            validate_func_annotations(
-                func,
-                has_args_override=expanded_args is not None,
-                has_returns_override=expanded_returns is not None,
-            )
-
-            if inspect.iscoroutinefunction(func):
-                async def async_wrapper(*args: Any, **kwargs: Any) -> UDFType:
-                    return await func(*args, **kwargs)  # type: ignore
-                async_wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
-                return functools.wraps(func)(async_wrapper)
+    def decorate(func: UDFType) -> UDFType:
+        if inspect.iscoroutinefunction(func):
+            async def async_wrapper(*args: Any, **kwargs: Any) -> UDFType:
+                return await func(*args, **kwargs)  # type: ignore
+            async_wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
+            wrapped: UDFType = functools.wraps(func)(async_wrapper)
 
-            else:
-                def wrapper(*args: Any, **kwargs: Any) -> UDFType:
-                    return func(*args, **kwargs)  # type: ignore
-                wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
-                return functools.wraps(func)(wrapper)
+        else:
+            def wrapper(*args: Any, **kwargs: Any) -> UDFType:
+                return func(*args, **kwargs)  # type: ignore
+            wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
+            wrapped = functools.wraps(func)(wrapper)
 
-        return decorate
+        # Build the signature now so that types the engine can't register
+        # fail at decoration time rather than when the DDL is executed.
+        get_signature(wrapped)
 
-    validate_func_annotations(
-        func,
-        has_args_override=expanded_args is not None,
-        has_returns_override=expanded_returns is not None,
-    )
+        return wrapped
 
-    if inspect.iscoroutinefunction(func):
-        async def async_wrapper(*args: Any, **kwargs: Any) -> UDFType:
-            return await func(*args, **kwargs)  # type: ignore
-        async_wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
-        return functools.wraps(func)(async_wrapper)
+    # No func was specified, this is an uncalled decorator that will get
+    # called later with the func to wrap.
+    if func is None:
+        return decorate
 
-    else:
-        def wrapper(*args: Any, **kwargs: Any) -> UDFType:
-            return func(*args, **kwargs)  # type: ignore
-        wrapper._singlestoredb_attrs = _singlestoredb_attrs  # type: ignore
-        return functools.wraps(func)(wrapper)
+    return decorate(func)
 
 
 def udf(
diff --git a/singlestoredb/functions/signature.py b/singlestoredb/functions/signature.py
index 290773df..3657ebcf 100644
--- a/singlestoredb/functions/signature.py
+++ b/singlestoredb/functions/signature.py
@@ -1185,6 +1185,48 @@ def vector_check(obj: Any) -> Tuple[Any, str]:
     return obj, 'scalar'
 
 
+def _is_dtype_str(annotation: str) -> bool:
+    """Is the string a dtype string (e.g. 'int64?', 'array[str]') not Python code?"""
+    dtype = annotation[:-1] if annotation.endswith('?') else annotation
+    dtype = dtype.replace('|null', '')
+    return (
+        dtype in sql_type_map
+        or dtype in unsupported_dtypes
+        or dtype.startswith(('array[', 'tuple['))
+    )
+
+
+def _resolve_annotation(annotation: Any, func: Callable[..., Any]) -> Any:
+    """
+    Evaluate a string annotation in the function's module namespace.
+
+    String annotations come from ``from __future__ import annotations`` or
+    quoted forward references. Dtype strings are left as-is.
+
+    """
+    if not isinstance(annotation, str) or _is_dtype_str(annotation):
+        return annotation
+    globalns = getattr(inspect.unwrap(func), '__globals__', {})
+    try:
+        return eval(annotation, globalns)
+    except Exception as exc:
+        raise TypeError(
+            f'could not resolve type annotation {annotation!r}: {exc}',
+        ) from exc
+
+
+def get_resolved_signature(func: Callable[..., Any]) -> inspect.Signature:
+    """Return the signature of ``func`` with string annotations evaluated."""
+    signature = inspect.signature(func)
+    return signature.replace(
+        parameters=[
+            p.replace(annotation=_resolve_annotation(p.annotation, func))
+            for p in signature.parameters.values()
+        ],
+        return_annotation=_resolve_annotation(signature.return_annotation, func),
+    )
+
+
 def get_masks(func: Callable[..., Any]) -> Tuple[List[bool], List[bool]]:
     """
     Get the list of masked parameters and return values for the function.
@@ -1202,8 +1244,9 @@ def get_masks(func: Callable[..., Any]) -> Tuple[List[bool], List[bool]]:
 
 
     """
-    params = inspect.signature(func).parameters
-    returns = inspect.signature(func).return_annotation
+    signature = get_resolved_signature(func)
+    params = signature.parameters
+    returns = signature.return_annotation
 
     ret_masks = []
     if typing.get_origin(returns) is Masked:
@@ -1242,7 +1285,7 @@ def get_signature(
     Dict[str, Any]
 
     '''
-    signature = inspect.signature(func)
+    signature = get_resolved_signature(func)
     args: List[Dict[str, Any]] = []
     returns: List[Dict[str, Any]] = []
 
@@ -1258,8 +1301,6 @@ def get_signature(
         elif p.kind == inspect.Parameter.VAR_KEYWORD:
             raise TypeError('variable keyword arguments are not supported')
 
-    # TODO: Use typing.get_type_hints() for parameters / return values?
-
     # Generate the parameter type and the corresponding SQL code for that parameter
     args_schema: List[ParamSpec] = []
     args_data_formats = []
@@ -1536,62 +1577,6 @@ def dtype_to_sql(
     return f'{sql_type}{nullable}{default_clause}'
 
 
-def _validate_annotation_types(annotation: Any) -> None:
-    """Validate that a parameter or return annotation contains no unsupported types."""
-    annotation = utils.resolve_type_alias(annotation)
-    if (
-        annotation is inspect.Parameter.empty
-        or annotation is None
-        or annotation is type(None)  # noqa: E721
-        or annotation is Any
-        or annotation is Ellipsis
-        or annotation in (list, tuple, dict)
-        or utils.is_dataframe(annotation)
-    ):
-        return
-
-    if isinstance(annotation, str):
-        dtype_to_sql(annotation)
-        return
-
-    if typing.get_origin(annotation) is not None:
-        for arg in typing.get_args(annotation):
-            _validate_annotation_types(arg)
-        return
-
-    if is_composite_type(annotation):
-        for pspec in get_colspec([annotation]):
-            _validate_annotation_types(pspec.dtype)
-        return
-
-    for item in simplify_dtype(annotation):
-        normalize_dtype(item)
-
-
-def validate_type_overrides(overrides: Optional[List[Any]]) -> None:
-    """Validate explicit args/returns type overrides passed to @udf."""
-    if not overrides:
-        return
-    for pspec in get_colspec(overrides):
-        _validate_annotation_types(pspec.dtype)
-
-
-def validate_func_annotations(
-    func: Callable[..., Any],
-    *,
-    has_args_override: bool = False,
-    has_returns_override: bool = False,
-) -> None:
-    """Validate parameter and return type annotations on a UDF callable."""
-    sig = inspect.signature(func)
-    if not has_args_override:
-        for param in sig.parameters.values():
-            if param.annotation is not inspect.Parameter.empty:
-                _validate_annotation_types(param.annotation)
-    if not has_returns_override and sig.return_annotation is not inspect.Signature.empty:
-        _validate_annotation_types(sig.return_annotation)
-
-
 def signature_to_sql(
     signature: Dict[str, Any],
     url: Optional[str] = None,
diff --git a/singlestoredb/tests/test_udf.py b/singlestoredb/tests/test_udf.py
index 59380f6b..d3ca8bfc 100755
--- a/singlestoredb/tests/test_udf.py
+++ b/singlestoredb/tests/test_udf.py
@@ -487,10 +487,9 @@ class TestUDF(unittest.TestCase):
         assert to_sql(foo) == '`foo`(`x` BIGINT NOT NULL) RETURNS SMALLINT NOT NULL'
 
         # Override multiple params with one type
-        @udf(args=dt.SMALLINT(nullable=False))
-        def foo(x: int, y: float, z: np.int8) -> int: ...
         with self.assertRaises(ValueError):
-            to_sql(foo)
+            @udf(args=dt.SMALLINT(nullable=False))
+            def foo(x: int, y: float, z: np.int8) -> int: ...
 
         # Override with list
         @udf(args=[dt.SMALLINT, dt.FLOAT, dt.CHAR(30)])
@@ -500,16 +499,14 @@ class TestUDF(unittest.TestCase):
             '`z` CHAR(30) NULL) RETURNS BIGINT NOT NULL'
 
         # Override with too short of a list
-        @udf(args=[dt.SMALLINT, dt.FLOAT])
-        def foo(x: int, y: float, z: str) -> int: ...
         with self.assertRaises(ValueError):
-            to_sql(foo)
+            @udf(args=[dt.SMALLINT, dt.FLOAT])
+            def foo(x: int, y: float, z: str) -> int: ...
 
         # Override with too long of a list
-        @udf(args=[dt.SMALLINT, dt.FLOAT, dt.CHAR(30), dt.TEXT])
-        def foo(x: int, y: float, z: str) -> int: ...
         with self.assertRaises(ValueError):
-            to_sql(foo)
+            @udf(args=[dt.SMALLINT, dt.FLOAT, dt.CHAR(30), dt.TEXT])
+            def foo(x: int, y: float, z: str) -> int: ...
 
         # Override with list
         @udf(args=[dt.SMALLINT, dt.FLOAT, dt.CHAR(30)])
diff --git a/singlestoredb/tests/test_udf_future_annotations.py b/singlestoredb/tests/test_udf_future_annotations.py
new file mode 100644
index 00000000..61b99bd5
--- /dev/null
+++ b/singlestoredb/tests/test_udf_future_annotations.py
@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+# type: ignore
+"""UDF signatures in a module using postponed evaluation of annotations."""
+from __future__ import annotations
+
+import datetime
+import unittest
+from typing import List
+from typing import NamedTuple
+from typing import Optional
+
+import numpy as np
+import numpy.typing as npt
+
+from ..functions import dtypes as dt
+from ..functions import Masked
+from ..functions import signature as sig
+from ..functions import Table
+from ..functions import udf
+
+
+def to_sql(x):
+    out = sig.signature_to_sql(sig.get_signature(x))
+    out = out.split('EXTERNAL FUNCTION ')[1]
+    return out.split('AS REMOTE')[0].strip()
+
+
+class Row(NamedTuple):
+    a: int
+    b: Optional[str]
+
+
+class TestFutureAnnotations(unittest.TestCase):
+
+    def test_scalars(self):
+        @udf
+        def foo(x: Optional[int], y: float = 1.5) -> Optional[str]: ...
+        assert to_sql(foo) == \
+            '`foo`(`x` BIGINT NULL, `y` DOUBLE NOT NULL DEFAULT 1.5e0) ' \
+            'RETURNS TEXT NULL'
+
+    def test_masked_vectors(self):
+        @udf
+        def foo(
+            x: npt.NDArray[np.float32],
+        ) -> Masked[npt.NDArray[np.float32]]: ...
+        assert to_sql(foo) == '`foo`(`x` FLOAT NOT NULL) RETURNS FLOAT NULL'
+
+    def test_table(self):
+        @udf
+        def foo(x: int) -> Table[List[Row]]: ...
+        assert to_sql(foo) == \
+            '`foo`(`x` BIGINT NOT NULL) ' \
+            'RETURNS TABLE(`a` BIGINT NOT NULL, `b` TEXT NULL)'
+
+    def test_dtype_strings(self):
+        @udf
+        def foo(x: 'str') -> 'array[int64]': ...  # noqa: F821
+        assert to_sql(foo) == \
+            '`foo`(`x` TEXT NOT NULL) RETURNS ARRAY(BIGINT NOT NULL) NOT NULL'
+
+    def test_unsupported_types(self):
+        with self.assertRaisesRegex(TypeError, 'datetime.date'):
+            @udf
+            def foo(x: datetime.date) -> int: ...
+
+        @udf(args=dt.TEXT, returns=dt.TEXT)
+        def bar(x: datetime.date) -> datetime.date: ...
+        assert to_sql(bar) == '`bar`(`x` TEXT NULL) RETURNS TEXT NULL'
+
+    def test_unresolvable(self):
+        class Local(NamedTuple):
+            a: int
+
+        with self.assertRaisesRegex(TypeError, 'could not resolve.*Local'):
+            @udf
+            def foo(x: int) -> Table[List[Local]]: ...
diff --git a/singlestoredb/tests/test_udf_returns.py b/singlestoredb/tests/test_udf_returns.py
index f1daa58d..b4457219 100644
--- a/singlestoredb/tests/test_udf_returns.py
+++ b/singlestoredb/tests/test_udf_returns.py
@@ -138,17 +138,10 @@ class UDFResultsTest(unittest.TestCase):
             '`foo_h_`(`x` TEXT NOT NULL) RETURNS TABLE(`a` TEXT NOT NULL)'
 
         # Plain TVF using multiple lists -- Illegal!
-        @udf
-        def foo_h(x: str) -> Table[List[int], List[str]]:
-            return Table(list(range(3)), [x] * 3)
-
-        foo_h_out = foo_h('cat')
-
-        assert type(foo_h_out) is Table
-        assert foo_h_out == Table([0, 1, 2], ['cat', 'cat', 'cat'])
-
         with self.assertRaises(TypeError):
-            to_sql(foo_h)
+            @udf
+            def foo_h(x: str) -> Table[List[int], List[str]]:
+                return Table(list(range(3)), [x] * 3)
 
         # Plain TVF using lists of NamedTuples
         @udf
@@ -361,16 +354,10 @@ class UDFResultsTest(unittest.TestCase):
         )
 
         # Vectorized TVF using lists -- Illegal!
-        @udf
-        def foo_t(x: List[str]) -> Table[List[int], List[str]]:
-            return Table(list(range(len(x))), x)
-
-        foo_t_out = foo_t(['cat', 'dog', 'monkey'])
-
-        assert type(foo_t_out) is Table
-        assert foo_t_out == Table([0, 1, 2], ['cat', 'dog', 'monkey'])
         with self.assertRaises(TypeError):
-            to_sql(foo_t)
+            @udf
+            def foo_t(x: List[str]) -> Table[List[int], List[str]]:
+                return Table(list(range(len(x))), x)
 
         # Vectorized TVF using pandas Series
         @udf(args=Parameters, returns=TVFTuple)

diff --git a/singlestoredb/functions/signature.py b/singlestoredb/functions/signature.py
index 0ce05a7e..2d4e8916 100644
--- a/singlestoredb/functions/signature.py
+++ b/singlestoredb/functions/signature.py
@@ -577,10 +577,11 @@ def get_dataclass_schema(obj: Any) -> List[ParamSpec]:
         A list of parameter specifications for the dataclass fields
 
     """
+    hints = utils.get_annotations(obj)
     return [
         ParamSpec(
             name=f.name,
-            dtype=f.type,
+            dtype=hints[f.name],
             default=NO_DEFAULT if f.default is dataclasses.MISSING else f.default,
         )
         for f in dataclasses.fields(obj)
diff --git a/singlestoredb/tests/test_udf_future_annotations.py b/singlestoredb/tests/test_udf_future_annotations.py
index 296540ef..98873974 100644
--- a/singlestoredb/tests/test_udf_future_annotations.py
+++ b/singlestoredb/tests/test_udf_future_annotations.py
@@ -3,6 +3,7 @@
 """UDF signatures in a module using postponed evaluation of annotations."""
 from __future__ import annotations
 
+import dataclasses
 import datetime
 import unittest
 from datetime import date
@@ -31,6 +32,12 @@ class Row(NamedTuple):
     b: Optional[str]
 
 
+@dataclasses.dataclass
+class DataclassRow:
+    a: int
+    b: Optional[str] = None
+
+
 class TestFutureAnnotations(unittest.TestCase):
 
     def test_scalars(self):
@@ -54,6 +61,13 @@ class TestFutureAnnotations(unittest.TestCase):
             '`foo`(`x` BIGINT NOT NULL) ' \
             'RETURNS TABLE(`a` BIGINT NOT NULL, `b` TEXT NULL)'
 
+    def test_dataclass_table(self):
+        @udf
+        def foo(x: int) -> Table[List[DataclassRow]]: ...
+        assert to_sql(foo) == \
+            '`foo`(`x` BIGINT NOT NULL) ' \
+            'RETURNS TABLE(`a` BIGINT NOT NULL, `b` TEXT NULL)'
+
     def test_dtype_strings(self):
         @udf
         def foo(x: 'str') -> 'array[int64]': ...  # noqa: F821